package com.atguigu.flinkTimer;

import com.atguigu.been.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import scala.Int;

import java.time.Duration;

/**
 * 	监控水位传感器的水位值，如果水位值在五秒钟之内(processing time)连续上升，
 * 	则报警，并将报警信息输出到侧输出流。
 * @author wky
 * @create 2021-07-18-18:03
 */
public class Timer_Test {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);
        DataStreamSource<String> streamSource = senv.socketTextStream("hadoop102", 9999);

        SingleOutputStreamOperator<WaterSensor> waterSensorDStream = streamSource.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
            }
        });
        WatermarkStrategy<WaterSensor> wms = WatermarkStrategy.<WaterSensor>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                        return element.getTs() * 1000;
                    }
                });
        SingleOutputStreamOperator<String> process = waterSensorDStream.assignTimestampsAndWatermarks(wms)
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    private Integer lastVc = Integer.MIN_VALUE;

                    //声明一个变量用来初始化定时器（保存定时器时间）
                    private Long timer = Long.MIN_VALUE;

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {


                    if (value.getVc()>lastVc){
                        if (lastVc==Integer.MIN_VALUE){
                            System.out.println("创建定时器");
                            timer = ctx.timestamp()+5000;
                            System.out.println("于"+(timer/1000+1)+"秒时报警");
//                            ctx.timerService().registerEventTimeTimer(ctx.timestamp()+5000);
                            ctx.timerService().registerEventTimeTimer(timer);
                            lastVc=value.getVc();
                        }
                    }
                    else {
                        System.out.println("删除定时器");
                        ctx.timerService().deleteEventTimeTimer(timer);
                        lastVc=Integer.MIN_VALUE;
                    }
                    out.collect(value.toString());
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        //System.out.println(ctx.timestamp()-5000);
                        ctx.output(new OutputTag<String>("up") {
                        },  (ctx.timestamp()/1000-5)+"到"+(ctx.timestamp()/1000)+"水位上涨");
                        lastVc = Integer.MIN_VALUE;
                    }
                });
        process.print("主流");
        process.getSideOutput(new OutputTag<String>("up"){}).print("报警信息");

        senv.execute();
    }
}
